package com.navinfo.tripanalysis.offline.service.impl.huashen;

import com.navinfo.tripanalysis.common.arithmetic.common.OuterEventData;
import com.navinfo.tripanalysis.common.arithmetic.enums.EventType;
import com.navinfo.tripanalysis.common.util.JsonUtils;
import com.navinfo.tripanalysis.offline.service.TripEventConvertService;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.expressions.GenericRow;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.bson.Document;

import java.util.ArrayList;
import java.util.List;


import com.navinfo.tripanalysis.common.pojo.RunningType;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;

/**
 * 行程事件信息的一汽转换服务
 * @author: web
 * @date: 2020-05-14
 **/
@RunningType(value = RunningTypeEnum.HS, interfaceClass = TripEventConvertService.class )
public class TripEventConvertServiceImpl implements TripEventConvertService {

    @Override
    public StructType createSchema() {
        List<StructField> list = new ArrayList<>();
        /*--start_schema*/
        list.add(new StructField("uid", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("tid", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("stype", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_gps_time", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("start_latitude", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_longitude", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_height", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_speed", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_gps_direction", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_gps_time", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("stop_latitude", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_longitude", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_height", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_speed", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_gps_direction", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("duration", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("event", DataTypes.StringType, true, Metadata.empty()));
        list.add(new StructField("extend_field01", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("extend_field02", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("extend_field03", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("extend_field04", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("extend_field05", DataTypes.DoubleType, true, Metadata.empty()));
        list.add(new StructField("extend_field06", DataTypes.DoubleType, true, Metadata.empty()));
        list.add(new StructField("extend_field07", DataTypes.DoubleType, true, Metadata.empty()));
        list.add(new StructField("extend_field08", DataTypes.DoubleType, true, Metadata.empty()));
        list.add(new StructField("extend_field09", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("extend_field10", DataTypes.LongType, true, Metadata.empty()));
        /*--end_schema*/

        return new StructType(list.toArray(new StructField[0]));
    }

    @Override
    public Row toRow(OuterEventData m, StructType schema) {
        List<Object> list = new ArrayList<>();
        /*--start_row*/
        list.add(m.getUid());
        list.add(m.getTid());
        list.add(m.getStype());
        list.add(m.getStartGpsTime());
        list.add(m.getStartLatitude());
        list.add(m.getStartLongitude());
        list.add(m.getStartHeight());
        list.add(m.getStartSpeed());
        list.add(m.getStartGpsDirection());
        list.add(m.getStopGpsTime());
        list.add(m.getStopLatitude());
        list.add(m.getStopLongitude());
        list.add(m.getStopHeight());
        list.add(m.getStopSpeed());
        list.add(m.getStopGpsDirection());
        list.add(m.getDuration());
        list.add(JsonUtils.toJsonStr(m.getEvent()));
        list.add(m.getExtendField01());
        list.add(m.getExtendField02());
        list.add(m.getExtendField03());
        list.add(m.getExtendField04());
        list.add(m.getExtendField05());
        list.add(m.getExtendField06());
        list.add(m.getExtendField07());
        list.add(m.getExtendField08());
        list.add(m.getExtendField09());
        list.add(m.getExtendField10());
        /*--end_row*/

        return null==schema ? new GenericRow(list.toArray()) : new GenericRowWithSchema(list.toArray(), schema);
    }

    @Override
    public OuterEventData fromRow(Row row) {
        OuterEventData m = new OuterEventData();
        /*--start_deserialization*/
        m.setUid(getLong(row,"uid"));
        m.setTid(getLong(row,"tid"));
        m.setStype(getInteger(row,"stype"));
        m.setStartGpsTime(getLong(row,"start_gps_time"));
        m.setStartLatitude(getInteger(row,"start_latitude"));
        m.setStartLongitude(getInteger(row,"start_longitude"));
        m.setStartHeight(getInteger(row,"start_height"));
        m.setStartSpeed(getInteger(row,"start_speed"));
        m.setStartGpsDirection(getInteger(row,"start_gps_direction"));
        m.setStopGpsTime(getLong(row,"stop_gps_time"));
        m.setStopLatitude(getInteger(row,"stop_latitude"));
        m.setStopLongitude(getInteger(row,"stop_longitude"));
        m.setStopHeight(getInteger(row,"stop_height"));
        m.setStopSpeed(getInteger(row,"stop_speed"));
        m.setStopGpsDirection(getInteger(row,"stop_gps_direction"));
        m.setDuration(getLong(row,"duration"));
        //ignore set setEvent
        m.setExtendField01(getInteger(row,"extend_field01"));
        m.setExtendField02(getInteger(row,"extend_field02"));
        m.setExtendField03(getInteger(row,"extend_field03"));
        m.setExtendField04(getInteger(row,"extend_field04"));
        m.setExtendField05(getDouble(row,"extend_field05"));
        m.setExtendField06(getDouble(row,"extend_field06"));
        m.setExtendField07(getDouble(row,"extend_field07"));
        m.setExtendField08(getDouble(row,"extend_field08"));
        m.setExtendField09(getLong(row,"extend_field09"));
        m.setExtendField10(getLong(row,"extend_field10"));
        /*--end_deserialization*/
        String eventName= getString(row,"event");
        m.setEvent(EventType.valueOf(eventName));
        m.setDay(getString(row,"day"));
        return m;
    }

    @Override
    public Document toDocument(OuterEventData m) {
        Document doc = new Document();
        /*--start_schema*/
        if(m.getUid()!=null){ doc.put("uid",m.getUid());}
        if(m.getTid()!=null){ doc.put("terminalId",m.getTid());}
        if(m.getStype()!=null){ doc.put("stype",m.getStype());}
        if(m.getStartGpsTime()!=null){ doc.put("startGpsTime",m.getStartGpsTime());}
        if(m.getStartLatitude()!=null){ doc.put("startLatitude",m.getStartLatitude());}
        if(m.getStartLongitude()!=null){ doc.put("startLongitude",m.getStartLongitude());}
        if(m.getStartHeight()!=null){ doc.put("startHeight",m.getStartHeight());}
        if(m.getStartSpeed()!=null){ doc.put("startSpeed",m.getStartSpeed());}
        if(m.getStartGpsDirection()!=null){ doc.put("startGpsDirection",m.getStartGpsDirection());}
        if(m.getStopGpsTime()!=null){ doc.put("stopGpsTime",m.getStopGpsTime());}
        if(m.getStopLatitude()!=null){ doc.put("stopLatitude",m.getStopLatitude());}
        if(m.getStopLongitude()!=null){ doc.put("stopLongitude",m.getStopLongitude());}
        if(m.getStopHeight()!=null){ doc.put("stopHeight",m.getStopHeight());}
        if(m.getStopSpeed()!=null){ doc.put("stopSpeed",m.getStopSpeed());}
        if(m.getStopGpsDirection()!=null){ doc.put("stopGpsDirection",m.getStopGpsDirection());}
        if(m.getDuration()!=null){ doc.put("duration",m.getDuration());}
//          EventType event -->
        if(m.getExtendField01()!=null){ doc.put("extendField01",m.getExtendField01());}
        if(m.getExtendField02()!=null){ doc.put("extendField02",m.getExtendField02());}
        if(m.getExtendField03()!=null){ doc.put("extendField03",m.getExtendField03());}
        if(m.getExtendField04()!=null){ doc.put("extendField04",m.getExtendField04());}
        if(m.getExtendField05()!=null){ doc.put("extendField05",m.getExtendField05());}
        if(m.getExtendField06()!=null){ doc.put("extendField06",m.getExtendField06());}
        if(m.getExtendField07()!=null){ doc.put("extendField07",m.getExtendField07());}
        if(m.getExtendField08()!=null){ doc.put("extendField08",m.getExtendField08());}
        if(m.getExtendField09()!=null){ doc.put("extendField09",m.getExtendField09());}
        if(m.getExtendField10()!=null){ doc.put("extendField10",m.getExtendField10());}
        /*--end_schema*/
        doc.put("day", m.getDay());
        if(m.getEvent()!=null){
            doc.put("event",m.getEvent().name());
        }
        return doc;
    }

    @Override
    public OuterEventData fromDocument(Document doc) {
        OuterEventData m = new OuterEventData();

        /*--start_row*/
        if(doc.containsKey("uid")){ m.setUid(doc.getLong("uid"));}
        if(doc.containsKey("terminalId")){ m.setTid(doc.getLong("terminalId"));}
        if(doc.containsKey("stype")){ m.setStype(doc.getInteger("stype"));}
        if(doc.containsKey("startGpsTime")){ m.setStartGpsTime(doc.getLong("startGpsTime"));}
        if(doc.containsKey("startLatitude")){ m.setStartLatitude(doc.getInteger("startLatitude"));}
        if(doc.containsKey("startLongitude")){ m.setStartLongitude(doc.getInteger("startLongitude"));}
        if(doc.containsKey("startHeight")){ m.setStartHeight(doc.getInteger("startHeight"));}
        if(doc.containsKey("startSpeed")){ m.setStartSpeed(doc.getInteger("startSpeed"));}
        if(doc.containsKey("startGpsDirection")){ m.setStartGpsDirection(doc.getInteger("startGpsDirection"));}
        if(doc.containsKey("stopGpsTime")){ m.setStopGpsTime(doc.getLong("stopGpsTime"));}
        if(doc.containsKey("stopLatitude")){ m.setStopLatitude(doc.getInteger("stopLatitude"));}
        if(doc.containsKey("stopLongitude")){ m.setStopLongitude(doc.getInteger("stopLongitude"));}
        if(doc.containsKey("stopHeight")){ m.setStopHeight(doc.getInteger("stopHeight"));}
        if(doc.containsKey("stopSpeed")){ m.setStopSpeed(doc.getInteger("stopSpeed"));}
        if(doc.containsKey("stopGpsDirection")){ m.setStopGpsDirection(doc.getInteger("stopGpsDirection"));}
        if(doc.containsKey("duration")){ m.setDuration(doc.getLong("duration"));}
//          EventType event -->
        if(doc.containsKey("extendField01")){ m.setExtendField01(doc.getInteger("extendField01"));}
        if(doc.containsKey("extendField02")){ m.setExtendField02(doc.getInteger("extendField02"));}
        if(doc.containsKey("extendField03")){ m.setExtendField03(doc.getInteger("extendField03"));}
        if(doc.containsKey("extendField04")){ m.setExtendField04(doc.getInteger("extendField04"));}
        if(doc.containsKey("extendField05")){ m.setExtendField05(doc.getDouble("extendField05"));}
        if(doc.containsKey("extendField06")){ m.setExtendField06(doc.getDouble("extendField06"));}
        if(doc.containsKey("extendField07")){ m.setExtendField07(doc.getDouble("extendField07"));}
        if(doc.containsKey("extendField08")){ m.setExtendField08(doc.getDouble("extendField08"));}
        if(doc.containsKey("extendField09")){ m.setExtendField09(doc.getLong("extendField09"));}
        if(doc.containsKey("extendField10")){ m.setExtendField10(doc.getLong("extendField10"));}
        /*--end_row*/
        m.setDay(doc.getString("day"));
        if(doc.containsKey("event")){
            String eventName=doc.getString("event");
            m.setEvent(EventType.valueOf(eventName));
        }
        return m;
    }

    @Override
    public Put toPut(OuterEventData m) {
        final byte[] d = Bytes.toBytes("d");
        long tid = m.getTid();
        Put put = new Put(Bytes.toBytes(new StringBuffer(String.valueOf(tid)).reverse().toString() + "|" + (3000000000000L - m.getStartGpsTime()) + "|" + m.getEvent().getNumber()));
        /*--start_schema*/
        if(m.getUid()!=null){put.addColumn(d, Bytes.toBytes("uid"), Bytes.toBytes(m.getUid()));}
        if(m.getTid()!=null){put.addColumn(d, Bytes.toBytes("terminalId"), Bytes.toBytes(m.getTid()));}
        if(m.getStype()!=null){put.addColumn(d, Bytes.toBytes("stype"), Bytes.toBytes(m.getStype()));}
        if(m.getStartGpsTime()!=null){put.addColumn(d, Bytes.toBytes("startGpsTime"), Bytes.toBytes(m.getStartGpsTime()));}
        if(m.getStartLatitude()!=null){put.addColumn(d, Bytes.toBytes("startLatitude"), Bytes.toBytes(m.getStartLatitude()));}
        if(m.getStartLongitude()!=null){put.addColumn(d, Bytes.toBytes("startLongitude"), Bytes.toBytes(m.getStartLongitude()));}
        if(m.getStartHeight()!=null){put.addColumn(d, Bytes.toBytes("startHeight"), Bytes.toBytes(m.getStartHeight()));}
        if(m.getStartSpeed()!=null){put.addColumn(d, Bytes.toBytes("startSpeed"), Bytes.toBytes(m.getStartSpeed()));}
        if(m.getStartGpsDirection()!=null){put.addColumn(d, Bytes.toBytes("startGpsDirection"), Bytes.toBytes(m.getStartGpsDirection()));}
        if(m.getStopGpsTime()!=null){put.addColumn(d, Bytes.toBytes("stopGpsTime"), Bytes.toBytes(m.getStopGpsTime()));}
        if(m.getStopLatitude()!=null){put.addColumn(d, Bytes.toBytes("stopLatitude"), Bytes.toBytes(m.getStopLatitude()));}
        if(m.getStopLongitude()!=null){put.addColumn(d, Bytes.toBytes("stopLongitude"), Bytes.toBytes(m.getStopLongitude()));}
        if(m.getStopHeight()!=null){put.addColumn(d, Bytes.toBytes("stopHeight"), Bytes.toBytes(m.getStopHeight()));}
        if(m.getStopSpeed()!=null){put.addColumn(d, Bytes.toBytes("stopSpeed"), Bytes.toBytes(m.getStopSpeed()));}
        if(m.getStopGpsDirection()!=null){put.addColumn(d, Bytes.toBytes("stopGpsDirection"), Bytes.toBytes(m.getStopGpsDirection()));}
        if(m.getDuration()!=null){put.addColumn(d, Bytes.toBytes("duration"), Bytes.toBytes(m.getDuration()));}
//          EventType event -->
        if(m.getExtendField01()!=null){put.addColumn(d, Bytes.toBytes("extendField01"), Bytes.toBytes(m.getExtendField01()));}
        if(m.getExtendField02()!=null){put.addColumn(d, Bytes.toBytes("extendField02"), Bytes.toBytes(m.getExtendField02()));}
        if(m.getExtendField03()!=null){put.addColumn(d, Bytes.toBytes("extendField03"), Bytes.toBytes(m.getExtendField03()));}
        if(m.getExtendField04()!=null){put.addColumn(d, Bytes.toBytes("extendField04"), Bytes.toBytes(m.getExtendField04()));}
        if(m.getExtendField05()!=null){put.addColumn(d, Bytes.toBytes("extendField05"), Bytes.toBytes(m.getExtendField05()));}
        if(m.getExtendField06()!=null){put.addColumn(d, Bytes.toBytes("extendField06"), Bytes.toBytes(m.getExtendField06()));}
        if(m.getExtendField07()!=null){put.addColumn(d, Bytes.toBytes("extendField07"), Bytes.toBytes(m.getExtendField07()));}
        if(m.getExtendField08()!=null){put.addColumn(d, Bytes.toBytes("extendField08"), Bytes.toBytes(m.getExtendField08()));}
        if(m.getExtendField09()!=null){put.addColumn(d, Bytes.toBytes("extendField09"), Bytes.toBytes(m.getExtendField09()));}
        if(m.getExtendField10()!=null){put.addColumn(d, Bytes.toBytes("extendField10"), Bytes.toBytes(m.getExtendField10()));}
        /*--end_schema*/
        if (m.getDay() != null) {
            put.addColumn(d, Bytes.toBytes("day"), Bytes.toBytes(m.getDay()));
        }
        if (m.getEvent() != null) {
            put.addColumn(d, Bytes.toBytes("event"), Bytes.toBytes(m.getEvent().getNumber()));
            put.addColumn(d, Bytes.toBytes("eventName"), Bytes.toBytes(m.getEvent().name()));
        }
        return put;
    }
}

